/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.quorum;

import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.server.ReplyCache;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.RequestProcessor;
import org.apache.zookeeper.server.SyncRequestProcessor;
import org.apache.zookeeper.server.paxos2.*;
import org.apache.zookeeper.server.ServerCnxn;
import org.apache.zookeeper.server.NIOServerCnxn;

import java.nio.ByteBuffer;

/**
 * This RequestProcessor simply forwards requests to an AckRequestProcessor and
 * SyncRequestProcessor.
 */
public class ProposalRequestProcessor implements RequestProcessor {
	LeaderZooKeeperServer zks;

	RequestProcessor nextProcessor;

	SyncRequestProcessor syncProcessor;

	public ProposalRequestProcessor(LeaderZooKeeperServer zks,
			RequestProcessor nextProcessor) {
		this.zks = zks;
		this.nextProcessor = nextProcessor;
		AckRequestProcessor ackProcessor = new AckRequestProcessor(zks
				.getLeader());
		syncProcessor = new SyncRequestProcessor(zks, ackProcessor);
	}

	/**
	 * initialize this processor
	 */
	public void initialize() {
		syncProcessor.start();
	}

	public void processRequest(Request[] request) {
		throw new RuntimeException("Not implemented");
	}

	public void processRequest(Request request) {
		// LOG.warn("Ack>>> cxid = " + request.cxid + " type = " +
		// request.type + " id = " + request.sessionId);
		// request.addRQRec(">prop");

		/*
		 * In the following IF-THEN-ELSE block, we process syncs on the leader.
		 * If the sync is coming from a follower, then the follower handler adds
		 * it to syncHandler. Otherwise, if it is a client of the leader that
		 * issued the sync command, then syncHandler won't contain the handler.
		 * In this case, we add it to syncHandler, and call processRequest on
		 * the next processor.
		 */
		// System.out.println("propose "+request);
        if(request == null){
            Leader.Proposal p = zks.getLeader().propose(request);

	    if (p != null) {
            	//nextProcessor.processRequest(p.request);
                syncProcessor.processRequest(p.request);
            }
	    zks.getLeader().queryLeader();
            return;
        }
        if (request.type == OpCode.prepareData) {
			syncProcessor.processRequest(new Request[] { request });
			return;
		}
		if (request instanceof LearnerSyncRequest) {
			zks.getLeader().processSync((LearnerSyncRequest) request);
		} else {
			if (request.type == OpCode.boloskyGetData) {
				if (Configuration.debug)
					System.out.println("Adding bolosky read to queue "
							+ Mapping.getBlockId(request));
				int size = 0;
				synchronized (zks.boloskyLock) {
					zks.bufferBoloskyReadQueue.addLast(request);
					size = zks.bufferBoloskyReadQueue.size();
				}
				//System.out.println(size+" "+Configuration.preferredReadBatchSize);
				if(size>=Configuration.preferredReadBatchSize)
					zks.getLeader().queryLeader();
				return;
			} // else
			// nextProcessor.processRequest(request);
			if (request.hdr != null) {
				// We need to sync and get consensus on any transactions
				if (request.cxid == ReplyCache
						.getLastOrdered(request.sessionId) + 1) {
					Leader.Proposal p = zks.getLeader().propose(request);
					if (p != null) {
				//		nextProcessor.processRequest(p.request);
						syncProcessor.processRequest(p.request);

					}
					ReplyCache.updateLastOrdered(request.sessionId,
							request.cxid);
				} else {
					System.out
							.println("Order aborted "
									+ request.cxid
									+ " expected "
									+ (ReplyCache
											.getLastOrdered(request.sessionId) + 1)
									+ " lastExecuted="
									+ ReplyCache.getLastCxid(request.sessionId));
					ByteBuffer reply = ReplyCache.getReply(request.sessionId,
							request.cxid);
					if (reply != null) {
						System.out.println("Retransmit " + request.sessionId
								+ " " + request.cxid);
						ServerCnxn cnxn = zks.getServerCnxnFactory()
								.getServerCnxn(request.sessionId);
						((NIOServerCnxn) cnxn).sendBuffer(reply);
					}
				}

			}
		}
	}

	public void shutdown() {
		nextProcessor.shutdown();
		syncProcessor.shutdown();
	}

}
